package com.example.websocket;

import android.content.Context;
import android.util.Log;

import java.util.List;
import java.util.concurrent.TimeUnit;

import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.X509TrustManager;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Timed;
import okhttp3.OkHttpClient;
import okio.ByteString;

public class RxWebSocket implements WebSocketWorker {
    private Context mContext;
    private final String TAG="RxWebSocket";
    /**
     * 是否打印Log
     */
    private boolean mIsPrintLog;
    /**
     * 支持外部传入OkHttpClient
     */
    private OkHttpClient mClient;
    /**
     * 支持SSL
     */
    private SSLSocketFactory mSslSocketFactory;
    private X509TrustManager mTrustManager;
    /**
     * 重连间隔时间
     */
    private long mReconnectInterval;
    /**
     * 重连间隔时间的单位
     */
    private TimeUnit mReconnectIntervalTimeUnit;
    /**
     * 具体干活的实现类
     */
    private WebSocketWorkerImpl mWorkerImpl;

    private RxWebSocket() {
    }

    public RxWebSocket(RxWebSocketBuilder builder) {
        this.mContext = builder.mContext;
        this.mIsPrintLog = builder.mIsPrintLog;
        this.mClient = builder.mClient == null ? new OkHttpClient() : builder.mClient;
        this.mSslSocketFactory = builder.mSslSocketFactory;
        this.mTrustManager = builder.mTrustManager;
        this.mReconnectInterval = builder.mReconnectInterval == 0 ? 1 : builder.mReconnectInterval;
        this.mReconnectIntervalTimeUnit = builder.mReconnectIntervalTimeUnit == null ? TimeUnit.SECONDS : builder.mReconnectIntervalTimeUnit;

        setup();
    }
    public boolean isConnect(String url){
        return mWorkerImpl.hasWebSocketConnection(url);
    }
    /**
     * 开始配置
     */
    private void setup() {
        this.mWorkerImpl = new WebSocketWorkerImpl(
                this.mContext,
                this.mIsPrintLog,
                this.mClient,
                this.mSslSocketFactory,
                this.mTrustManager,
                this.mReconnectInterval,
                this.mReconnectIntervalTimeUnit);
    }

    public Observable<Boolean> heartBeat(String url, int period, TimeUnit unit,
                                         HeartBeatGenerateCallback heartBeatGenerateCallback) {
        if (heartBeatGenerateCallback == null) {
            return Observable.error(new NullPointerException("heartBeatGenerateCallback == null"));
        }
        return Observable
                .interval(period, unit)
                //timestamp操作符，给每个事件加一个时间戳
                .timestamp()
                .retry()
                .flatMap(new Function<Timed<Long>, ObservableSource<Boolean>>() {
                    @Override
                    public ObservableSource<Boolean> apply(Timed<Long> timed) throws Exception {
                        long timestamp = timed.time();
                        //判断网络，存在网络才发消息，否则直接返回发送心跳失败
                        if (mContext != null && NetworkUtil.hasNetWorkStatus(mContext, false)) {
                            String heartBeatMsg = heartBeatGenerateCallback.onGenerateHeartBeatMsg(timestamp);
                            Log.d(TAG,  "发送心跳消息: " + heartBeatMsg);
                            if (mWorkerImpl.hasWebSocketConnection(url)) {
                                return send(url, heartBeatMsg);
                            } else {
                                //这里必须用异步发送，如果切断网络，再重连，缓存的WebSocket会被清除，此时再重连网络
                                //是没有WebSocket连接可用的，所以就需要异步连接完成后，再发送
                                return asyncSend(url, heartBeatMsg);
                            }
                        } else {
                            Log.d(TAG, "无网络连接，不发送心跳，下次网络连通时，再次发送心跳");
                            return Observable.create(new ObservableOnSubscribe<Boolean>() {
                                @Override
                                public void subscribe(ObservableEmitter<Boolean> emitter) throws Exception {
                                    emitter.onNext(false);
                                }
                            });
                        }
                    }
                });
    }
    @Override
    public Observable<WebSocketInfo> get(String url) {
        return mWorkerImpl.get(url);
    }

    @Override
    public Observable<WebSocketInfo> get(String url, long timeout, TimeUnit timeUnit) {
        return mWorkerImpl.get(url,timeout,timeUnit);
    }

    @Override
    public Observable<Boolean> send(String url, String msg) {
        return mWorkerImpl.send(url,msg);
    }

    @Override
    public Observable<Boolean> send(String url, ByteString byteString) {
        return mWorkerImpl.send(url,byteString);
    }

    @Override
    public Observable<Boolean> asyncSend(String url, String msg) {
        return mWorkerImpl.asyncSend(url,msg);
    }

    @Override
    public Observable<Boolean> asyncSend(String url, ByteString byteString) {
        return mWorkerImpl.asyncSend(url,byteString);
    }

    @Override
    public Observable<Boolean> close(String url) {
        return mWorkerImpl.close(url);
    }

    @Override
    public boolean closeNow(String url) {
        return mWorkerImpl.closeNow(url);
    }

    @Override
    public Observable<List<Boolean>> closeAll() {
        return mWorkerImpl.closeAll();
    }

    @Override
    public void closeAllNow() {
        mWorkerImpl.closeAllNow();
    }
}